package com.gcl.rabbitmq.utils.com.gcl.rabbitmq;

import com.gcl.rabbitmq.utils.RabbitMQUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.TimeoutException;

public class AsyncMessageConfirm {


    public static final String QUEUE_NAME = "Confirm";


    public static void main(String[] args) throws IOException, TimeoutException {

        Channel channel = RabbitMQUtil.getChannel();

        // 开启发布确认
        channel.confirmSelect();


        ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();

        // 成功的回调
        ConfirmCallback ackConfirmCallback = (deliveryTag, multiple) ->{
            System.out.println("确认的消息 "+ deliveryTag);
            outstandingConfirms.remove(deliveryTag);
        };

        // 失败的回调
        /**
         * 1. deliveryTag消息的标识
         * 2. multiple 是否为批量
         */
        ConfirmCallback nackConfirmCallback = (deliveryTag, multiple) ->{
            System.out.println("未确认的消息 "+ deliveryTag);
        };


        // 开启消息的监听器
        channel.addConfirmListener(ackConfirmCallback,nackConfirmCallback);


        // 声明队列
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        // 绑定路由

        // 获取消息

        for (int i = 0; i < 1000; i++) {
            String msg = i + "";
            channel.basicPublish("",QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,msg.getBytes(StandardCharsets.UTF_8));
            outstandingConfirms.put(channel.getNextPublishSeqNo(), msg);
        }

    }
}
